/*
 * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 */

/*
 *
 *
 *
 *
 *
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/publicdomain/zero/1.0/
 */

package java.util.concurrent;
import java.util.Collection;
import java.util.concurrent.locks.LockSupport;

/**
 * A cancellable asynchronous computation.  This class provides a base
 * implementation of {@link Future}, with methods to start and cancel
 * a computation, query to see if the computation is complete, and
 * retrieve the result of the computation.  The result can only be
 * retrieved when the computation has completed; the {@code get}
 * methods will block if the computation has not yet completed.  Once
 * the computation has completed, the computation cannot be restarted
 * or cancelled (unless the computation is invoked using
 * {@link #runAndReset}).
 *
 * <p>A {@code FutureTask} can be used to wrap a {@link Callable} or
 * {@link Runnable} object.  Because {@code FutureTask} implements
 * {@code Runnable}, a {@code FutureTask} can be submitted to an
 * {@link Executor} for execution.
 *
 * <p>In addition to serving as a standalone class, this class provides
 * {@code protected} functionality that may be useful when creating
 * customized task classes.
 *
 * 【核心方法】：{@link #run()}
 *
 * Q1：FutureTask 是怎样在 run() 方法中获取返回值的？
 * Q2：它把返回值放到哪里了？
 * Q3：get() 方法又是怎样拿到这个返回值的呢？
 *
 * @since 1.5
 * @author Doug Lea
 * @param <V> The result type returned by this FutureTask's {@code get} methods
 */
public class FutureTask<V> implements RunnableFuture<V> {
    /*
     * Revision notes: This differs from previous versions of this
     * class that relied on AbstractQueuedSynchronizer, mainly to
     * avoid surprising users about retaining interrupt status during
     * cancellation races. Sync control in the current design relies
     * on a "state" field updated via CAS to track completion, along
     * with a simple Treiber stack to hold waiting threads.
     *
     * Style note: As usual, we bypass overhead of using
     * AtomicXFieldUpdaters and instead directly use Unsafe intrinsics.
     */

    /**
     * The run state of this task, initially NEW.  The run state
     * transitions to a terminal state only in methods set,
     * setException, and cancel.  During completion, state may take on
     * transient values of COMPLETING (while outcome is being set) or
     * INTERRUPTING (only while interrupting the runner to satisfy a
     * cancel(true)). Transitions from these intermediate to final
     * states use cheaper ordered/lazy writes because values are unique
     * and cannot be further modified.
     *
     * <pre>
     * state: 当前任务的运行状态，初始值为new。
     * Possible state transitions（可能存在的状态转换）:
     *      NEW -> COMPLETING -> NORMAL  // 执行过程顺利完成，参见方法 {@link #set(Object)}
     *      NEW -> COMPLETING -> EXCEPTIONAL // 执行过程出现异常，参见方法 {@link #setException(Throwable)}
     *      NEW -> CANCELLED // 执行过程被取消
     *      NEW -> INTERRUPTING -> INTERRUPTED // 执行过程中，线程被中断
     * </pre>
     */
    private volatile int state;
    private static final int NEW          = 0; // 初始化
    private static final int COMPLETING   = 1; // 执行过程中，任务已经完成，但还没有将执行结果存储到outcome
    private static final int NORMAL       = 2; // 任务正常执行完成，并且将执行结果已经存储到outcome
    private static final int EXCEPTIONAL  = 3; // 执行过程中，任务出现异常，并且将异常信息已经存储到outcome
    private static final int CANCELLED    = 4; // 任务被取消
    private static final int INTERRUPTING = 5; // 线程被中断
    private static final int INTERRUPTED  = 6;

    /** The underlying callable; nulled out after running */
    private Callable<V> callable;
    /**
     * The result to return or exception to throw from get()
     * <pre>
     * outcome的赋值在{@link #run()} 方法中：
     *  outcome用来存储 {@link Callable#call()} 方法返回的【执行结果】或【异常信息】。
     *  通过 {@link #set(Object)} 和 {@link #setException(Throwable)} 这两个方法设置 outcome。
     * </pre>
     */
    private Object outcome; // non-volatile, protected by state reads/writes
    /**
     * The thread running the callable; CASed during run()
     * 执行这个任务的线程。
     */
    private volatile Thread runner;
    /** Treiber stack of waiting threads */
    private volatile WaitNode waiters;

    /**
     * Returns result or throws exception for completed task.
     * 返回已完成任务的结果（{@link #run()} 方法中调用 {@link Callable#call()} 方法返回的【执行结果】） 或 异常
     *
     * @param s completed state value
     */
    @SuppressWarnings("unchecked")
    private V report(int s) throws ExecutionException {
        Object x = outcome;
        if (s == NORMAL)
            return (V)x;
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable)x);
    }

    /**
     * Creates a {@code FutureTask} that will, upon running, execute the
     * given {@code Callable}.
     *
     * 构造函数的入参很重要，这就是 FutureTask 可以拿到异步任务执行结果的原因所在，
     * 具体的直接看 {@link #run()} 方法就知道了。
     *
     * @param  callable the callable task
     * @throws NullPointerException if the callable is null
     */
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }

    /**
     * Creates a {@code FutureTask} that will, upon running, execute the
     * given {@code Runnable}, and arrange that {@code get} will return the
     * given result on successful completion.
     *
     * @param runnable the runnable task
     * @param result the result to return on successful completion. If
     * you don't need a particular result, consider using
     * constructions of the form:
     * {@code Future<?> f = new FutureTask<Void>(runnable, null)}
     * @throws NullPointerException if the runnable is null
     */
    public FutureTask(Runnable runnable, V result) {
        // 入参虽然是 Runnable，但是在这里还是需要转换为 Callable。
        // 因为毕竟 Runnable 是拿不到异步任务的执行结果的，而 Callable 就可以拿到。
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }

    // 判断任务是否被取消
    public boolean isCancelled() {
        return state >= CANCELLED;
    }

    public boolean isDone() {
        return state != NEW;
    }

    /**
     * <pre>
     * 取消任务的执行。
     *
     * 1. 执行 cancel 操作一定失败的情况：
     *    ① 任务已经执行完成；
     *    ② 任务已经被取消过了；
     *    ③ 任务因为某种原因不能被取消。
     *    其他情况下，cancel 操作都会返回 true。
     *
     * 2. 该方法真的会停止异步任务的执行吗？
     *       FutureTask.cancel(true)会尝试中断 执行任务的线程（调用该线程的 interrupt()中断方法），
     *    但任务是否真正停止取决于任务代码是否对中断做出响应，比如，没有 主动检查中断标志 或 未处理中断异常 任务就不会真正的停止。
     *       也就是说，如果任务代码未处理中断，即使调用 cancel(true)，任务仍会继续运行。
     *
     *    我们可以看看 {@link Thread#interrupt()}的方法签名有如下解释：
     *        根据 Java的中断机制，只有 当线程处于 等待状态/超时等待状（比如，wait()、sleep(long)方法）或者 主动检查中断标志 时，中断才会【真正的生效】。
     *     也就是说，其他线程 调用了 当前线程的interrupt()方法后（将中断标志设置为true），如果 当前线程 没有 主动检查中断标志
     *     或者 当前线程线程 没有处于 waiting/timed_waiting状态，那么也就意味着 当前线程 并不会真正的停止任务执行。
     *
     *  3. 入参mayInterruptIfRunning
     *     如果发起 cancel 的任务已经处于运行状态，那么依据 mayInterruptIfRunning 的参数值有两种处理方式：
     *      ① 如果为 true，则当前在执行的任务会被中断。
     *      ② 如果为 false，则可以允许正在执行的任务继续运行，直到它执行完成。
     *  不管入参为 true还是false，{@link #get()}方法：在返回执行结果的时候，会根据 任务的运行状态state 进行一定的判断，如果是 CANCELLED，则会抛出CancellationException异常。
     * @param mayInterruptIfRunning
     * </pre>
     */
    public boolean cancel(boolean mayInterruptIfRunning) {
        // 如果 任务状态不是new 或 CAS失败（设置 state[任务的运行状态] 失败），则返回 false，表示任务取消失败。
        // mayInterruptIfRunning 为 true，则将任务状态设置为 INTERRUPTING，表示任务【正在被中断】；
        // mayInterruptIfRunning 为 false，则将任务状态设置为 CANCELLED，表示任务已经取消。
        if (!(state == NEW &&
              UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            return false;
        try {    // in case call to interrupt throws exception
            // 需要中断任务执行线程
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner; // runnerOffset
                    if (t != null)
                        // 中断线程
                        t.interrupt();
                } finally { // final state
                    // 将【任务的运行状态 state】设置为 interrupted（已中断）
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }
        } finally {
            // 唤醒等待队列中的线程
            finishCompletion();
        }
        return true;
    }

    /**
     * <pre>
     * 【阻塞】等待任务完成并返回结果，从 awaitDone() 方法就可以看出来。
     *
     * 获取 {@link #run()} 方法中调用 {@link Callable#call()} 方法返回的【执行结果】或者【异常】等信息。
     *
     * 如果程序运行正常的话，通常调用 get() 方法，会将当前线程挂起。
     * Q：那么，挂起的线程由谁来唤醒呢？
     * A：自然是 {@link #run()} 方法运行完后唤醒。
     *    在设置返回结果 {@link #set(Object)} 或者 设置异常 {@link #setException(Throwable)} 的两个方法中都会调用，
     *    {@link #finishCompletion()} 方法，这个方法就会唤醒等待队列中的线程。
     * </pre>
     * @throws CancellationException {@inheritDoc}
     */
    public V get() throws InterruptedException, ExecutionException {
        int s = state;
        /**
         * 1- 如果【任务的运行状态 state <= COMPLETING】为true，则说明任务还【未】执行完成，
         *    此时，就需要调用 awaitDone() 方法【阻塞】 调用get()方法的当前线程，直到得到执行结果（正常的结果或者异常信息）。
         * @see AbstractExecutorService#invokeAll(Collection) ，看看这个方法是怎么使用 FutureTask 的。
         *
         *    从{@link #run()}方法中就可以看出来：
         *      ① run方法在正常执行完任务后会调用{@link #set(Object)}方法修改同步状态为completing，并将执行结果存放到outcome中。
         *      ② run方法在执行任务时，如果发生了异常，则会调用{@link #setException(Throwable)}方法修改同步状态为completing，并将异常信息存放到outcome中。
         */
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        // 2- 返回结果
        return report(s);
    }

    /**
     * @throws CancellationException {@inheritDoc}
     */
    public V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        if (unit == null)
            throw new NullPointerException();
        int s = state;
        if (s <= COMPLETING &&
            (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
            throw new TimeoutException();
        return report(s);
    }

    /**
     * Protected method invoked when this task transitions to state
     * {@code isDone} (whether normally or via cancellation).
     * The default implementation does nothing. 默认实现什么也不做。
     * Subclasses may override this method to invoke completion callbacks or perform bookkeeping.
     * 子类可以重写该方法，用来执行一些回调或者进行一些记录。
     * Note that you can query status inside the
     * implementation of this method to determine whether this task
     * has been cancelled.
     */
    protected void done() { }

    /**
     * Sets the result of this future to the given value unless
     * this future has already been set or has been cancelled.
     *
     * <p>This method is invoked internally by the {@link #run} method
     * upon successful completion of the computation.
     *
     * 保存 {@link #run()} 方法中调用的 {@link Callable#call()} 方法返回的【执行结果】。
     *
     * @param v the value
     */
    protected void set(V v) {
        // 通过 CAS 的方式，将 state（任务的运行状态）由 new 更新为 completing
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            // 将 Callable#call() 方法返回的【执行结果】保存到 outcome 中。
            outcome = v;
            // 最终将 state（任务的运行状态）由 completing 更新为 normal
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }

    /**
     * Causes this future to report an {@link ExecutionException}
     * with the given throwable as its cause, unless this future has
     * already been set or has been cancelled.
     *
     * <p>This method is invoked internally by the {@link #run} method
     * upon failure of the computation.
     *
     * 保存 {@link #run()} 方法中调用的 {@link Callable#call()} 方法返回的【异常信息】。
     *
     * @param t the cause of failure
     */
    protected void setException(Throwable t) {
        // 通过 CAS 的方式，将 state（任务的运行状态）由 new 更新为 completing
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            // 将 Callable#call() 方法返回的【异常信息】保存到 outcome 中。
            outcome = t;
            // 最终将 state（任务的运行状态）由 completing 状态更新为 exceptional 状态。
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
            finishCompletion();
        }
    }

    /**
     * <pre>
     * 执行任务，这个方法会被 【执行线程】调用。
     *  FutureTask 调用 {@link Callable#call()} 方法获得执行结果，并将其存起来，然后 {@link #get()} 方法就能获取到。
     *      ① 异步任务正常执行结束，会把执行结果存放到 {@link #outcome} 中，看方法 {@link #set(Object)}；
     *      ② 异步任务执行异常，会把抛出的异常信息也存放到 {@link #outcome} 中，看方法 {@link #setException(Throwable)}。
     * </pre>
     */
    public void run() {
        /**
         * ① 如果状态不是 new，说明任务已经执行或者已经被取消了，直接返回。
         * ② 如果状态是 new，则尝试【把 执行线程信息 保存在 runnerOffset（{@link #runner} 字段）】; 如果赋值失败，则直接返回。
         */
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            // 获取构造函数传入的 Callable 值
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    /**
                     * 开始执行任务，正常调用 {@link Callable#call()} 就可以获取返回值。
                     * 这里调用的任务
                     *  ①要么是程序员自己实现的Callable#call()方法;
                     *  ②要么就是 {@link FutureTask#FutureTask(Runnable, Object)} 里面默认实现的Callable#call()方法。
                     */
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    // 保存 call 方法抛出的异常
                    setException(ex);
                }
                if (ran)
                    // 保存 call 方法的执行结果
                    set(result);
            }
        } finally {
            // runner must be non-null until state is settled to
            // prevent concurrent calls to run()
            runner = null;
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            int s = state;
            // 如果任务被中断，则执行中断处理
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }

    /**
     * Executes the computation without setting its result, and then
     * resets this future to initial state, failing to do so if the
     * computation encounters an exception or is cancelled.  This is
     * designed for use with tasks that intrinsically execute more
     * than once.
     *
     * @return {@code true} if successfully run and reset
     */
    protected boolean runAndReset() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return false;
        boolean ran = false;
        int s = state;
        try {
            Callable<V> c = callable;
            if (c != null && s == NEW) {
                try {
                    c.call(); // don't set result
                    ran = true;
                } catch (Throwable ex) {
                    setException(ex);
                }
            }
        } finally {
            // runner must be non-null until state is settled to
            // prevent concurrent calls to run()
            runner = null;
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
        return ran && s == NEW;
    }

    /**
     * Ensures that any interrupt from a possible cancel(true) is only
     * delivered to a task while in run or runAndReset.
     */
    private void handlePossibleCancellationInterrupt(int s) {
        // It is possible for our interrupter to stall before getting a
        // chance to interrupt us.  Let's spin-wait patiently.
        if (s == INTERRUPTING)
            while (state == INTERRUPTING)
                Thread.yield(); // wait out pending interrupt

        // assert state == INTERRUPTED;

        // We want to clear any interrupt we may have received from
        // cancel(true).  However, it is permissible to use interrupts
        // as an independent mechanism for a task to communicate with
        // its caller, and there is no way to clear only the
        // cancellation interrupt.
        //
        // Thread.interrupted();
    }

    /**
     * Simple linked list nodes to record waiting threads in a Treiber stack.
     * See other classes such as Phaser and SynchronousQueue for more detailed explanation.
     * WaitNode 用来记录 等待获取结果的线程。具体的看 {@link #awaitDone(boolean, long)}方法。
     */
    static final class WaitNode {
        volatile Thread thread;
        volatile WaitNode next;
        WaitNode() { thread = Thread.currentThread(); }
    }

    /**
     * Removes and signals all waiting threads, invokes done(), and
     * nulls out callable.
     */
    private void finishCompletion() {
        // assert state > COMPLETING;
        for (WaitNode q; (q = waiters) != null;) {
            if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
                for (;;) {
                    Thread t = q.thread;
                    if (t != null) {
                        q.thread = null;
                        // 唤醒等待队列中的线程
                        LockSupport.unpark(t);
                    }
                    WaitNode next = q.next;
                    if (next == null)
                        break;
                    q.next = null; // unlink to help gc
                    q = next;
                }
                break;
            }
        }

        /**
         * 默认实现什么也不做。子类可以重写该方法，在任务执行完成后，用来执行一些回调或者进行一些记录。
         * 这里很重要，在 {@link ExecutorCompletionService#completionQueue} 中有体现。
         */
        done();

        callable = null;        // to reduce footprint
    }

    /**
     * Awaits completion or aborts on interrupt or timeout.
     * 有等待就会有队列排队 或者 可响应中断，从方法签名上看有 InterruptedException，那就说明该方法时可以被中断的。
     *
     * @param timed true if use timed waits
     * @param nanos time to wait, if timed
     * @return state upon completion
     */
    private int awaitDone(boolean timed, long nanos)
        throws InterruptedException {
        // 计算等待截止时间
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        WaitNode q = null;
        boolean queued = false;

        /**
         * 自旋：这个自旋【通常】会经历三次循环。
         * 1). 第一轮 for 循环，执行的逻辑是 q == null，这个时候会新建一个结点 q，第一轮循环结束。
         * 2). 第二轮 for 循环，执行的逻辑是 !queue，这个时候会把第一轮循环中生成的结点的 next 指针指向
         *     waiters，然后通过 CAS 的方式用结点 q 替换 waiters，也就是把新生成的结点添加到 waiters 中的首结点。
         *     如果替换成功，则 queue = true。第二轮循环结束。
         * 3). 第三轮 for 循环，进行阻塞等待。要么等待特定时间，要么一直阻塞，直到被其他线程唤醒。
         */
        for (;;) {
            // 1- 判断当前线程是否被中断。如果是，则在等待队列中删除该结点，接着抛出 InterruptedException
            if (Thread.interrupted()) {
                removeWaiter(q);
                throw new InterruptedException();
            }

            // 2-
            // 获取 任务的运行状态
            int s = state;
            // 2-1 任务的运行状态 > completing（中间状态），说明已经达到了某个最终状态（normal正常/exceptional异常/cancelled取消）
            if (s > COMPLETING) {
                if (q != null)
                    // 将 thread 置为 null
                    q.thread = null;
                return s;
            }
            // 2-2 任务的运行状态 为 completing（完成状态），表示任务已经执行完成（即 Callable#call() 方法已经执行完成），
            //    只是 outcome 还没有赋值，也就是说，set()方法 或者 setException()方法 还没执行完成。
            else if (s == COMPLETING) // cannot time out yet
                // 主动让出执行权，让其他线程优先执行（只是发出了这个信号，至于别的线程是否会一定执行就不晓得了）
                Thread.yield();
            else if (q == null)
                // 将当前线程构造结点
                q = new WaitNode();
            else if (!queued)
                // 如果还没有入队，则把当前结点加入 waiters 首结点（q.next = waiters），并替换原来的 waiters。
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                     q.next = waiters, q);
            // 如果有超时时间
            else if (timed) {
                // 计算还剩多少时间
                nanos = deadline - System.nanoTime();
                if (nanos <= 0L) {
                    removeWaiter(q);
                    return state;
                }
                // 阻塞等待 特定时间
                LockSupport.parkNanos(this, nanos);
            }
            else
                /**
                 * 挂起当前线程，直到被其他线程唤醒。
                 *
                 * {@link #finishCompletion()} 中会唤醒线程。
                 */
                LockSupport.park(this);
        }
    }

    /**
     * Tries to unlink a timed-out or interrupted wait node to avoid
     * accumulating garbage.  Internal nodes are simply unspliced
     * without CAS since it is harmless if they are traversed anyway
     * by releasers.  To avoid effects of unsplicing from already
     * removed nodes, the list is retraversed in case of an apparent
     * race.  This is slow when there are a lot of nodes, but we don't
     * expect lists to be long enough to outweigh higher-overhead
     * schemes.
     */
    private void removeWaiter(WaitNode node) {
        if (node != null) {
            node.thread = null;
            retry:
            for (;;) {          // restart on removeWaiter race
                for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
                    s = q.next;
                    if (q.thread != null)
                        pred = q;
                    else if (pred != null) {
                        pred.next = s;
                        if (pred.thread == null) // check for race
                            continue retry;
                    }
                    else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                          q, s))
                        continue retry;
                }
                break;
            }
        }
    }

    // Unsafe mechanics
    private static final sun.misc.Unsafe UNSAFE;
    private static final long stateOffset;
    private static final long runnerOffset;
    private static final long waitersOffset;
    static {
        try {
            UNSAFE = sun.misc.Unsafe.getUnsafe();
            Class<?> k = FutureTask.class;
            stateOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("state"));
            runnerOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("runner"));
            waitersOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("waiters"));
        } catch (Exception e) {
            throw new Error(e);
        }
    }

}
